package com.li.spark0615.zhibiao

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType

object Eleven {
  def main(args: Array[String]): Unit = {

    System.setProperty("HADOOP_USER_NAME","root")

    val session = SparkSession
      .builder()
      .appName("li")
      .master("local[*]")
      .enableHiveSupport()
      .config("hive.metastore.uris", "thrift://192.168.23.40:9083")
      .config("dfs.client.use.datanode.hostname", "true")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .getOrCreate()

    val frame = session
      .table("ods.lx_job").limit(1000)
      .as("a")
      .join(session.table("ods.lx_area").as("b"), col("a.city") === col("b.city"), "left")
      .select(
        //        col("a.id"), col("a.comid"), col("a.title"), col("state"), concat(col("money"), lit("_"), col("maney_max")) as "money"
        col("a.id") , col("a.comid") , col("a.title") , col("state") , col("name") as "cityname" , col("money") , col("maney_max").cast(DoubleType) as "maney_max"
      )

    frame.show()


    val value = frame
      .distinct()
      .as("a")
      .join(
        session.table("ods.lx_com").as("b")
        ,
        col("a.comid") === col("b.id")
      )
      .selectExpr("a.title","a.state as area","a.money","a.cityname","a.maney_max","b.name")
      .where(col("area") === "北京市")
      .orderBy(col("maney_max") desc)
      .groupBy(col("area"),col("cityname"))
      .agg(
      first("title") as "title",
      first("name") as "name",
      first("money") as "money",
        max(col("maney_max")) as "maney_max"
    )
      .select(
                col("title"), col("name"), col("cityname"), col("area"), concat(col("money"), lit("_"), col("maney_max")) as "money"
//        col("title") , col("name") , col("cityname") , col("state") , col("name") as "cityname" , col("money") , col("maney_max").cast(DoubleType) as "maney_max"
      )

    value
      //      .groupBy("")
      .write
      .format("jdbc")
      .option("url", "jdbc:mysql://192.168.23.40:3306/shtd_store?useSSL=false&characterEncoding=utf8")
      .option("user", "root")
      .option("password", "123456")
      .option("dbtable", "col_11")
      .mode(SaveMode.Overwrite)
      .save()


  }
}
